#include <iostream>
#include <iomanip>
#include <string>
#include <memory>
#include <chrono>
#include <thread>

#include "./utility/socket/xdgTcpServer.hpp" // TCP服务器头文件
#include "./utility/threadPool/xdgThreadPool.hpp" // 线程池头文件
#include "./utility/config/inicpp.h"          // ini配置文件
#include "./utility/config/nlohmannjson.hpp"  // json文件
#include "./utility/log/xdgspdlog.hpp"     // 日志文件
#include "./utility/database/DBManager.hpp"      // 数据库头文件
#include "./utility/client/clientManger.hpp"     // 客户端信息、接口管理头文件
#include "./utility/videoCodec/StreamBus.hpp"    // 视频流处理头文件

xdg::ThreadPool g_pool;                  // 全局线程池实例
std::atomic<bool> g_streamStop{false};   // 视频流频流停止标志
void onError(const std::string &errMsg); // 错误回调函数前向声明


// 广播推流
inline void pullAllUsers(std::shared_ptr<std::vector<char>> buf)
{
    std::vector<int> users = UsersPull::getInstance().getUsers();
    if (users.empty()) return;

    std::string packet;                             // 创建一个空的数据包
    packet.reserve(4 + buf->size());            // 预留容量，避免反复扩容
    uint32_t body_len = buf->size();               // 1. 获取消息体长度
    packet.append(reinterpret_cast<const char *>(&body_len), 4); // 2. 先将4字节的长度头拼接到包中
    packet.append(buf->begin(), buf->end());                             // 3. 再将消息体本身拼接到包中

    for (int fd : users)
    {
        xdg::sendProxy::instance().send(fd, packet.data(), packet.size());
    }
}


// ======================= 回调函数实现 =======================

// 具体处理数据的函数（在线程池中运行）
void handleMessage(std::shared_ptr<xdg::TcpContext> context, std::shared_ptr<std::vector<char>> pack)
{

    std::string msg(pack->begin() + 4, pack->end()); // 去掉头部的4字节长度，获取消息体
    nlohmann::json json_data;
    try
    {
        json_data = nlohmann::json::parse(msg);
    }
    catch (const nlohmann::json::parse_error &e)
    {
        std::string str = "handleMessage JSON解析错误: ";
        str += e.what();
        onError(str);
        return;
    }

    int type = json_data["type"]; // 请求类型

    std::string ProcessingResult;
    switch (type)
    {
    case RequestType::REQUEST_GETSALT:
        LOG_CONSOLE_INFO("收到客户端 获取盐值 请求，请求类型为：{}，IP地址：{}", type, context->clientIp());
        ProcessingResult = interface_getSalt(context, json_data["data"]["account"]);
        break;
    case RequestType::REQUEST_REGISTER:
        LOG_CONSOLE_INFO("收到客户端 注册 请求，请求类型为：{}，IP地址：{}", type, context->clientIp());
        ProcessingResult = interface_register(context, json_data["data"]["account"], json_data["data"]["password"]);
        break;
    case RequestType::REQUEST_LOGIN:
        LOG_CONSOLE_INFO("收到客户端 登录 请求，请求类型为：{}，IP地址：{}", type, context->clientIp());
        ProcessingResult = interface_login(context, json_data["data"]["account"], json_data["data"]["password"]);
        break;
    case RequestType::REQUEST_JOINRTSP:
        LOG_CONSOLE_INFO("收到客户端 RTSP 请求，请求类型为：{}，IP地址：{}", type, context->clientIp());
        ProcessingResult = interface_addpull(context, json_data["data"]["account"]);
        LOG_CONSOLE_INFO("当前用户数：{}", UsersPull::getInstance().getUserCount());
        break;
    case RequestType::REQUEST_QUITRTSP:
        LOG_CONSOLE_INFO("收到客户端 RTSP 退出请求，请求类型为：{}，IP地址：{}", type, context->clientIp());
        interface_delpull(context);
        break;

    default:
        std::string str = "handleMessage 未知请求类型: ";
        str += std::to_string(type);
        onError(str);
        ProcessingResult = "{\"type\": 404, \"success\": false, \"message\": \"404\"}";
        return;
    }

    LOG_CONSOLE_INFO("请求处理完成，请求类型为：{}，IP地址：{}", type, context->clientIp());

    // 消息体格式：4字节数据长度 + 4字节json长度 + json消息体 + 其他（可省略，数据长度放到json消息体中）
    std::vector<char> packet;
    packet.reserve(8 + ProcessingResult.length());     // 预留容量，避免反复扩容
    uint32_t body_len = ProcessingResult.length() + 4; // 1. 总数据的长度
    uint32_t json_len = ProcessingResult.length();     // 2. json的长度
    
    // 3. 先将4字节总长度头拼接到包中
    packet.insert(packet.begin(), 
    reinterpret_cast<const char *>(&body_len), reinterpret_cast<const char*>(&body_len) + sizeof(body_len));
    // 4. 再将4字节json长度头拼接到包中
    packet.insert(packet.begin() + 4, 
    reinterpret_cast<const char *>(&json_len), reinterpret_cast<const char*>(&json_len) + sizeof(json_len));
    // 5. 再将json本身拼接到包中
    packet.insert(packet.begin() + 8, ProcessingResult.begin(), ProcessingResult.end());

    context->send(packet.data(), packet.size());
    return;
}

// 新客户端连接回调
int onConnection(std::shared_ptr<xdg::TcpContext> context)
{
    LOG_CONSOLE_INFO("[+] 接受新用户连接: {}:{}", context->clientIp(), context->clientPort());
    struct UserInfo user;
    user.isLogin = false;
    user.username = "";
    user.pwd_hash = "";
    user.salt = "";
    user.ip = context->clientIp();
    user.port = context->clientPort();

    UsersInfoManager::getInstance().addUser(context->getFd(), user);
    return 0; // 返回0表示接受连接
}

// 数据到达（完整数据包）回调
void onMessage(std::shared_ptr<xdg::TcpContext> context, std::shared_ptr<std::vector<char>> pack)
{
    try
    {
        g_pool.submit(handleMessage, context, pack);
    }
    catch (const std::exception &e)
    {
        std::string str = "onMessage submit error：";
        str = str + e.what();
        onError(str);
    }

    return;
}

// 客户端断开连接回调
void onClose(std::shared_ptr<xdg::TcpContext> context)
{
    LOG_CONSOLE_INFO("[-] 用户断开连接: {}:{}", context->clientIp(), context->clientPort());
    UsersInfoManager::getInstance().delUser(context->getFd());
}

// 服务器错误回调
void onError(const std::string &errMsg)
{
    LOG_CONSOLE_ERROR("[!] Server Error: {}", errMsg);
}



// ======================= 主函数 =======================
int main(void)
{
    xdg_spdlog_init(); // 初始化日志系统

    ini::IniFile ini;
    ini.load("./server.ini"); // 载入配置文件

    // ================================== 视频流初始化 ==================================
    StreamBus::Instance().InitFFmpegNetwork(); // 全局初始化FFmpeg网络

    // 打开源并设置“入队策略”（在 StreamBus 内部的回调里深拷贝）
    if (!StreamBus::Instance().Open(ini["Stream"]["stream_url"].as<std::string>()))
    {
        LOG_CONSOLE_INFO("打开视频流失败");
        return 1;
    }
    LOG_CONSOLE_INFO("打开视频流成功");

    // 创建一个线程，用于从 StreamBus 中读取数据包并发送
    std::thread senderHandle([](){
        while(!g_streamStop.load()){
            std::shared_ptr<PacketBlob> b;
            if ( !StreamBus::Instance().WaitDequeue(b)  || !b  ) continue;  // 获取失败或者数据包为空则跳过

            std::shared_ptr<std::vector<char>> buf = BuildFrameChunkBuffer(*b);
            if (!buf || buf->empty()) { LOG_CONSOLE_WARN("数据包为空");  continue; } // 数据包为空则跳过
            pullAllUsers(buf); // 推送给所有推流用户
        }

    });

    
    StreamBus::Instance().Start(); // 启动读线程：内部循环 ReadNextPacket()，回调里快速“拷贝+入队”


    // ================================== 数据库初始化 ==================================
    LOG_CONSOLE_INFO("开始初始化数据库！");
    try
    {
        DBManager::init(DBManager::createMySqlConnectString(
            ini["Database"]["host"].as<std::string>(),
            ini["Database"]["port"].as<int>(),
            ini["Database"]["database_name"].as<std::string>(),
            ini["Database"]["username"].as<std::string>(),
            ini["Database"]["password"].as<std::string>()));
        LOG_CONSOLE_INFO("数据库连接池初始化成功！");
    }
    catch (const std::exception &e)
    {
        LOG_CONSOLE_ERROR("初始化失败 (Ini Error)，请检查数据库配置文件: {}", e.what());
        return 1;
    }

    // ================================== 服务器初始化 ==================================
    std::string SERVER_IP = ini["Server"]["listen_ip"].as<std::string>(); // 监听所有网络接口
    int SERVER_PORT = ini["Server"]["listen_port"].as<int>();
    if (SERVER_IP.empty())
    {
        LOG_CONSOLE_WARN("listen_ip is empty, use 0.0.0.0");
        SERVER_IP = "0.0.0.0";
    }
    if (SERVER_PORT < 1 || SERVER_PORT > 65535)
    {
        LOG_CONSOLE_WARN("listen_port is invalid, use 8080");
        SERVER_PORT = 8080;
    }
    ini["Server"]["listen_ip"] = SERVER_IP; // 写监听地址配置项
    ini["Server"]["listen_port"] = SERVER_PORT;
    // ini.save("./server.ini"); 保存配置文件

    CONSOLE_LOG_INFO("------ TCP Echo Server ------");

    // 1. 创建服务器实例
    xdg::TcpServer server(SERVER_IP, SERVER_PORT);
    g_pool.start(); // 启动线程池

    // 2. 设置拆包规则（4字节头部(消息体长度) + 消息体）
    server.setUnpackingRule(xdg::UnpackingRule::HEADER_BODY, 102400);

    // 3. 绑定回调函数
    server.setClientConnectCallback(onConnection);  // 新客户端连接回调
    server.setDataArrivedCallbackShared(onMessage); // 数据到达（完整数据包）回调
    server.setClientDisconnectCallback(onClose);    // 客户端断开连接回调
    server.setErrorCallback(onError);               // 服务器错误回调

    // 4. 启动服务器
    if (server.start() != 0)
    {
        CONSOLE_LOG_ERROR("Failed to start server.");
        return 1;
    }

    LOG_CONSOLE_INFO("[OK] TCP 回显服务器启动成功，监听端口：{}:{}", SERVER_IP, SERVER_PORT);
    CONSOLE_LOG_INFO("输入 'q' 或 'quit' 或 '回车' 后按回车键来停止服务器...");

    // 5. 主线程阻塞，等待用户输入 "quit"
    std::string cmd;
    while (std::getline(std::cin, cmd))
    {
        if (cmd == "q" || cmd == "quit" || cmd.empty())
        {
            break;
        }
    }

    // 6. 停止所有操作，回收资源

    // 停止拉流+清理
    g_streamStop.store(true); // 停止 StreamBus
    senderHandle.join();

    StreamBus::Instance().Stop(/*drain=*/true);
    StreamBus::Instance().Close();
    StreamBus::Instance().DeinitFFmpegNetwork();

    // 停止服务器
    CONSOLE_LOG_INFO("正在停止服务器...");
    server.stop();
    CONSOLE_LOG_INFO("[OK] 服务器已停止.");

    g_pool.destroy(); // 停止线程池
    xdg_spdlog_shutdown();

    return 0;
}


